//go:build mqtt
// +build mqtt

package connector

import (
	"context"
	"fmt"
	"sync"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
)

// MQTTSourceConnector represents an MQTT source connector
type MQTTSourceConnector struct {
	// config is the connector configuration
	config map[string]interface{}
	// client is the MQTT client
	client mqtt.Client
	// metrics are the connector metrics
	metrics map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
	// started indicates if the connector is started
	started bool
	// topic is the MQTT topic
	topic string
	// broker is the MQTT broker
	broker string
	// clientID is the MQTT client ID
	clientID string
	// username is the MQTT username
	username string
	// password is the MQTT password
	password string
	// qos is the MQTT QoS
	qos byte
	// messages is the message channel
	messages chan mqtt.Message
	// buffer is the message buffer
	buffer []mqtt.Message
	// batchSize is the batch size
	batchSize int
}

// NewMQTTSourceConnector creates a new MQTT source connector
func NewMQTTSourceConnector() SourceConnector {
	return &MQTTSourceConnector{
		metrics:   make(map[string]interface{}),
		qos:       0,
		messages:  make(chan mqtt.Message, 100),
		buffer:    make([]mqtt.Message, 0),
		batchSize: 10,
	}
}

// Initialize initializes the connector
func (c *MQTTSourceConnector) Initialize(config map[string]interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Store the configuration
	c.config = config

	// Get topic
	if topic, ok := config["topic"].(string); ok {
		c.topic = topic
	} else {
		return fmt.Errorf("topic is required")
	}

	// Get broker
	if broker, ok := config["broker"].(string); ok {
		c.broker = broker
	} else {
		return fmt.Errorf("broker is required")
	}

	// Get client ID
	if clientID, ok := config["client_id"].(string); ok {
		c.clientID = clientID
	} else {
		c.clientID = fmt.Sprintf("dataflare-mqtt-source-%d", time.Now().UnixNano())
	}

	// Get username
	if username, ok := config["username"].(string); ok {
		c.username = username
	}

	// Get password
	if password, ok := config["password"].(string); ok {
		c.password = password
	}

	// Get QoS
	if qos, ok := config["qos"].(float64); ok {
		c.qos = byte(qos)
	}

	// Get batch size
	if batchSize, ok := config["batch_size"].(float64); ok {
		c.batchSize = int(batchSize)
	}

	// Initialize metrics
	c.metrics["records_total"] = 0
	c.metrics["records_read"] = 0
	c.metrics["start_time"] = time.Now().Unix()

	return nil
}

// Start starts the connector
func (c *MQTTSourceConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already started
	if c.started {
		return fmt.Errorf("connector already started")
	}

	// Create MQTT options
	opts := mqtt.NewClientOptions()
	opts.AddBroker(c.broker)
	opts.SetClientID(c.clientID)
	if c.username != "" {
		opts.SetUsername(c.username)
	}
	if c.password != "" {
		opts.SetPassword(c.password)
	}
	opts.SetDefaultPublishHandler(c.messageHandler)
	opts.SetAutoReconnect(true)
	opts.SetCleanSession(true)

	// Create client
	client := mqtt.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		return fmt.Errorf("failed to connect to MQTT broker: %w", token.Error())
	}
	c.client = client

	// Subscribe to topic
	if token := client.Subscribe(c.topic, c.qos, nil); token.Wait() && token.Error() != nil {
		return fmt.Errorf("failed to subscribe to MQTT topic: %w", token.Error())
	}

	c.started = true

	return nil
}

// Stop stops the connector
func (c *MQTTSourceConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already stopped
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Unsubscribe from topic
	if token := c.client.Unsubscribe(c.topic); token.Wait() && token.Error() != nil {
		return fmt.Errorf("failed to unsubscribe from MQTT topic: %w", token.Error())
	}

	// Disconnect client
	c.client.Disconnect(250)

	c.started = false

	return nil
}

// Read reads records from the source
func (c *MQTTSourceConnector) Read(ctx context.Context) (*RecordBatch, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return nil, fmt.Errorf("connector not started")
	}

	// Check if we have buffered messages
	if len(c.buffer) > 0 {
		// Get batch size
		batchSize := c.batchSize
		if batchSize > len(c.buffer) {
			batchSize = len(c.buffer)
		}

		// Get messages
		messages := c.buffer[:batchSize]
		c.buffer = c.buffer[batchSize:]

		// Convert messages to records
		records := make([]*Record, len(messages))
		for i, message := range messages {
			record := NewRecord([]byte(message.Topic()), message.Payload())
			record.Timestamp = time.Now()
			records[i] = record
		}

		// Update metrics
		c.metrics["records_read"] = c.metrics["records_read"].(int) + len(records)
		c.metrics["last_read_time"] = time.Now().Unix()

		return NewRecordBatch(records), nil
	}

	// Read messages from channel
	for {
		select {
		case message := <-c.messages:
			// Add message to buffer
			c.buffer = append(c.buffer, message)
			// Update metrics
			c.metrics["records_total"] = c.metrics["records_total"].(int) + 1
			// Check if we have enough messages
			if len(c.buffer) >= c.batchSize {
				// Get messages
				messages := c.buffer[:c.batchSize]
				c.buffer = c.buffer[c.batchSize:]

				// Convert messages to records
				records := make([]*Record, len(messages))
				for i, message := range messages {
					record := NewRecord([]byte(message.Topic()), message.Payload())
					record.Timestamp = time.Now()
					records[i] = record
				}

				// Update metrics
				c.metrics["records_read"] = c.metrics["records_read"].(int) + len(records)
				c.metrics["last_read_time"] = time.Now().Unix()

				return NewRecordBatch(records), nil
			}
		case <-time.After(100 * time.Millisecond):
			// No message available
			if len(c.buffer) > 0 {
				// Get messages
				messages := c.buffer
				c.buffer = make([]mqtt.Message, 0)

				// Convert messages to records
				records := make([]*Record, len(messages))
				for i, message := range messages {
					record := NewRecord([]byte(message.Topic()), message.Payload())
					record.Timestamp = time.Now()
					records[i] = record
				}

				// Update metrics
				c.metrics["records_read"] = c.metrics["records_read"].(int) + len(records)
				c.metrics["last_read_time"] = time.Now().Unix()

				return NewRecordBatch(records), nil
			}
			return nil, nil
		}
	}
}

// Commit commits the offset
func (c *MQTTSourceConnector) Commit(ctx context.Context, offset interface{}) error {
	// MQTT doesn't support commit
	return nil
}

// GetMetrics gets connector metrics
func (c *MQTTSourceConnector) GetMetrics() map[string]interface{} {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy metrics
	metrics := make(map[string]interface{})
	for k, v := range c.metrics {
		metrics[k] = v
	}

	return metrics
}

// messageHandler handles MQTT messages
func (c *MQTTSourceConnector) messageHandler(client mqtt.Client, message mqtt.Message) {
	// Add message to channel
	c.messages <- message
}

// MQTTSinkConnector represents an MQTT sink connector
type MQTTSinkConnector struct {
	// config is the connector configuration
	config map[string]interface{}
	// client is the MQTT client
	client mqtt.Client
	// metrics are the connector metrics
	metrics map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
	// started indicates if the connector is started
	started bool
	// topic is the MQTT topic
	topic string
	// broker is the MQTT broker
	broker string
	// clientID is the MQTT client ID
	clientID string
	// username is the MQTT username
	username string
	// password is the MQTT password
	password string
	// qos is the MQTT QoS
	qos byte
	// retained indicates if messages should be retained
	retained bool
}

// NewMQTTSinkConnector creates a new MQTT sink connector
func NewMQTTSinkConnector() SinkConnector {
	return &MQTTSinkConnector{
		metrics:  make(map[string]interface{}),
		qos:      0,
		retained: false,
	}
}

// Initialize initializes the connector
func (c *MQTTSinkConnector) Initialize(config map[string]interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Store the configuration
	c.config = config

	// Get topic
	if topic, ok := config["topic"].(string); ok {
		c.topic = topic
	} else {
		return fmt.Errorf("topic is required")
	}

	// Get broker
	if broker, ok := config["broker"].(string); ok {
		c.broker = broker
	} else {
		return fmt.Errorf("broker is required")
	}

	// Get client ID
	if clientID, ok := config["client_id"].(string); ok {
		c.clientID = clientID
	} else {
		c.clientID = fmt.Sprintf("dataflare-mqtt-sink-%d", time.Now().UnixNano())
	}

	// Get username
	if username, ok := config["username"].(string); ok {
		c.username = username
	}

	// Get password
	if password, ok := config["password"].(string); ok {
		c.password = password
	}

	// Get QoS
	if qos, ok := config["qos"].(float64); ok {
		c.qos = byte(qos)
	}

	// Get retained
	if retained, ok := config["retained"].(bool); ok {
		c.retained = retained
	}

	// Initialize metrics
	c.metrics["records_total"] = 0
	c.metrics["start_time"] = time.Now().Unix()

	return nil
}

// Start starts the connector
func (c *MQTTSinkConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already started
	if c.started {
		return fmt.Errorf("connector already started")
	}

	// Create MQTT options
	opts := mqtt.NewClientOptions()
	opts.AddBroker(c.broker)
	opts.SetClientID(c.clientID)
	if c.username != "" {
		opts.SetUsername(c.username)
	}
	if c.password != "" {
		opts.SetPassword(c.password)
	}
	opts.SetAutoReconnect(true)
	opts.SetCleanSession(true)

	// Create client
	client := mqtt.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		return fmt.Errorf("failed to connect to MQTT broker: %w", token.Error())
	}
	c.client = client

	c.started = true

	return nil
}

// Stop stops the connector
func (c *MQTTSinkConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already stopped
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Disconnect client
	c.client.Disconnect(250)

	c.started = false

	return nil
}

// Write writes records to the sink
func (c *MQTTSinkConnector) Write(ctx context.Context, batch *RecordBatch) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Check if batch is valid
	if batch == nil || len(batch.Records) == 0 {
		return nil
	}

	// Write records
	for _, record := range batch.Records {
		// Get topic
		topic := c.topic
		if len(record.Key) > 0 {
			// Use record key as topic suffix
			topic = fmt.Sprintf("%s/%s", topic, string(record.Key))
		}

		// Publish message
		token := c.client.Publish(topic, c.qos, c.retained, record.Value)
		if token.Wait() && token.Error() != nil {
			return fmt.Errorf("failed to publish message: %w", token.Error())
		}
	}

	// Update metrics
	c.metrics["records_total"] = c.metrics["records_total"].(int) + len(batch.Records)
	c.metrics["last_write_time"] = time.Now().Unix()

	return nil
}

// Flush flushes any buffered records
func (c *MQTTSinkConnector) Flush(ctx context.Context) error {
	// MQTT client doesn't buffer messages
	return nil
}

// GetMetrics gets connector metrics
func (c *MQTTSinkConnector) GetMetrics() map[string]interface{} {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy metrics
	metrics := make(map[string]interface{})
	for k, v := range c.metrics {
		metrics[k] = v
	}

	return metrics
}
